package com.moli.iov.base.kafka;

import com.alibaba.fastjson.JSONObject;
import com.moli.iov.base.remote.IVehicleBaseRemoteService;
import com.moli.iov.consumer.KafkaConsumerServer;
import com.moli.iov.protocol.model.dto.VehicleOnLineDto;
import lombok.extern.slf4j.Slf4j;
import org.apache.dubbo.config.annotation.Reference;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

/**
 *
 * Online remind consumer message listener.
 *
 * @author OAK
 * @since 2020/02/27 16:44:00 PM.
 *
 */
@Slf4j
public class OnOffConsumerMessageListener extends KafkaConsumerServer {

    @Autowired
    private ThreadPoolTaskExecutor taskExecutor;

    @Reference(group = "${service.group}", check = false, timeout = 10000)
    private IVehicleBaseRemoteService iVehicleBaseRemoteService;

    @Override
    public void manualHandle(ConsumerRecord<String, Object> record,
                             Acknowledgment acknowledgment) {
        try{
            onlineHandle(record.topic(), (VehicleOnLineDto)record.value());
        }catch(Exception e){
            log.error("kafka 消费者 车辆在线提醒 添加线程池异常", e);
        }finally{
            acknowledgment.acknowledge();
        }
    }

    private void onlineHandle(String topic,
                              VehicleOnLineDto vehicleOnLineDto){
        taskExecutor.execute(() -> {
            try {
                if(vehicleOnLineDto == null){
                    return;
                }
                log.info("Kafka 消费者 车辆在线提醒 线程池中处理开始 接收到消息  topic: {} 数据：{}",
                        topic, JSONObject.toJSON(vehicleOnLineDto));

                if(vehicleOnLineDto == null){
                    log.error("Kafka 消费者 车辆在线提醒 topic: {} 数据：{}",
                            topic, vehicleOnLineDto);
                    return;
                }

                if(vehicleOnLineDto.getStatus()){
                    iVehicleBaseRemoteService.onlineRemind(vehicleOnLineDto.getVin());
                }
            }catch(Exception e){
                log.error("aKafka 消费者 车辆在线提醒 处理异常", e);
            }
        });
    }
}
